package com.szsky.sky.service.mq;

import com.szsky.sky.dto.mq.ToZeroProducerDto;
import com.szsky.sky.producer.UumToZeroProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Service
public class AsyncMessageService {
    private static final Logger log = LoggerFactory.getLogger(AsyncMessageService.class);
    
    @Autowired
    private UumToZeroProducer uumToZeroProducer;
    
    /**
     * 异步启动消息发送任务
     * @Async 注解确保该方法在独立线程中执行
     */
    @Async
    public void startAsyncSending(List<ToZeroProducerDto> dataList) {
        int totalCount = dataList.size();
        log.info("开始后台发送任务，共需发送 {} 条消息，每1000毫秒1条", totalCount);
        
        // 创建单线程调度池
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(
            r -> {
                Thread thread = new Thread(r);
                thread.setName("background-message-sender");
                return thread;
            }
        );
        
        AtomicInteger sentCount = new AtomicInteger(0);
        
        // 定时任务：每秒发送一条
        scheduler.scheduleAtFixedRate(() -> {
            try {
                int currentIndex = sentCount.getAndIncrement();
                
                // 所有消息发送完毕，关闭线程池
                if (currentIndex >= totalCount) {
                    log.info("全部 {} 条消息发送完成，关闭线程池", totalCount);
                    scheduler.shutdown();
                    return;
                }
                
                // 发送当前消息
                ToZeroProducerDto toZeroProducerDto = dataList.get(currentIndex);
                uumToZeroProducer.sendSyncMessage(toZeroProducerDto); // 调用你的消息生产者
                
                // 打印进度（每100条打印一次）
                if ((currentIndex + 1) % 100 == 0) {
                    log.info("已发送 {}/{} 条消息", currentIndex + 1, totalCount);
                }
            } catch (Exception e) {
                log.error("发送第 {} 条消息失败", sentCount.get(), e);
                // 可根据需要添加重试逻辑
            }
        }, 0, 1000, TimeUnit.MILLISECONDS); // 每1000毫秒发送一次
        
        // 监控线程池关闭
        monitorScheduler(scheduler, totalCount);
    }
    
    /**
     * 监控线程池状态，确保资源正确释放
     */
    private void monitorScheduler(ScheduledExecutorService scheduler, int totalCount) {
        new Thread(() -> {
            try {
                // 等待线程池完成所有任务
                if (scheduler.awaitTermination(1, TimeUnit.HOURS)) { // 超时时间设置为1小时
                    log.info("线程池正常关闭，所有消息发送完成");
                } else {
                    log.warn("发送超时，强制关闭线程池");
                    scheduler.shutdownNow();
                }
            } catch (InterruptedException e) {
                log.error("线程池监控被中断", e);
                scheduler.shutdownNow();
            }
        }, "scheduler-monitor").start();
    }
}
    